package mr.stocks.question1;

import java.io.IOException;

import mr.stocks.model.StocksInputWritable;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.LineRecordReader;
import au.com.bytecode.opencsv.CSVParser;

public class StocksWritablesRecordReader extends RecordReader<LongWritable, StocksInputWritable>{
    private final LineRecordReader lineReader = new LineRecordReader();
    private final StocksInputWritable stocks = new StocksInputWritable();
    private CSVParser csvParser = new CSVParser();
    @Override
    public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
        lineReader.initialize(split, context);
        lineReader.nextKeyValue();
    }

    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {
        boolean hasNextKey = lineReader.nextKeyValue();
        if (hasNextKey){
            parse(lineReader.getCurrentValue());
        }
        return hasNextKey;
    }
    
    private void parse(Text value) throws IOException {
    	
    	String[] split = csvParser.parseLine(value.toString());
        stocks.setStockSymbol(split[0]);
        stocks.setTimeStamp(Long.parseLong(split[1]));
        stocks.setPrice(Double.parseDouble(split[2]));
    }

    @Override
    public LongWritable getCurrentKey() throws IOException, InterruptedException {
        return lineReader.getCurrentKey();
    }

    @Override
    public StocksInputWritable getCurrentValue() throws IOException, InterruptedException {
        return stocks;
    }

    @Override
    public float getProgress() throws IOException, InterruptedException {
        return lineReader.getProgress();
    }

    @Override
    public void close() throws IOException {
        lineReader.close();
    }
}